package studio.raptor.databus.client;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.cmdb.Config;
import studio.raptor.cmdb.ConfigService;

import javax.jms.JMSException;
import javax.jms.Session;

/**
 * 队列消息生产者，每个生产者只能发送消息到一个指定的主题
 * <p>
 * 默认配置如下:<br>
 * 发送时间默认不超时<br>
 * failover尝试重连次数一直重连<br>
 * failover两次重连之间最大时间为10s<br>
 * 开启事务性会话<br>
 * 使用自动ACK模式<br>
 * </p>
 */
public class DatabusProducerBuilder {

  private static Logger log = LoggerFactory.getLogger(DatabusProducerBuilder.class);

  /**
   * 发送操作的超时时间，此时间并不会因重连操作打断。-1表示不超时。
   */
  private int timeout = -1;
  /**
   * 两次重连之间最长时间间隔，10s.
   */
  private int maxReconnectDelay = 10000;

  /**
   * failover机制中的最大重连次数。-1表示一直重连。
   */
  private int maxReconnectAttempts = -1;
  /**
   * 客户端启动时最大重连次数，默认为5
   */
  private int startupMaxReconnectAttempts = 5;

  private String otherURLConfig = "&";

  /**
   * 会话是否为事务性，生产者默认开启
   */
  private boolean transactedSession = true;

  /**
   * 消息ACK模式，默认值(1)为自动确认
   */
  private int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;

  /**
   * 消息大小设置，单位 KB.若批量发送，则表示所有的事件的大小之和
   */
  private int eventMaxSize = 1024*5;


  /**
   * 发送消息的有效期，单位天
   */
  private int msgTimeToLive = 7;


  private DatabusConfig databusConfig;

  public DatabusProducerBuilder(DatabusConfig databusConfig) {
    this.databusConfig = databusConfig;
  }


  /**
   * 创建生产者
   */
  public DatabusProducer create() throws JMSException {
    DatabusProducer producer
        = new DatabusProducer(databusConfig, this, this.databusConfig.getBrokerGroups());
    //初始化生产者配置
    try {
      initConfig();
    } catch (IOException e) {
      log.error(e.getMessage());
      throw new JMSException(e.getMessage());
    }
    return producer;
  }

  /**
   * 依据配置加载方式初始化生产者配置
   */
  private void initConfig() throws IOException {
    this.setAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
    if (this.databusConfig.getLoadType() == DatabusConfigLoadType.LOAD_TYPE_CMDB) {
      Config config = ConfigService
          .getConfig(DatabusConfig.CLIENT_CONFIG_PRE + databusConfig.getBizKey());
      this.setStartupMaxReconnectAttempts(config
          .getIntProperty("producer.startupMaxReconnectAttempts",
              this.getStartupMaxReconnectAttempts()));
      this.setMaxReconnectAttempts(
          config.getIntProperty("producer.maxReconnectAttempts", this.getMaxReconnectAttempts()));
      this.setOtherURLConfig(
          config.getProperty("producer.otherURLConfig", this.getOtherURLConfig()));
      this.setTimeout(config.getIntProperty("producer.timeout", this.getTimeout()));
      this.setTransactedSession(
          config.getBooleanProperty("producer.transactedSession", this.isTransactedSession()));
      this.setMaxReconnectDelay(
          config.getIntProperty("producer.maxReconnectDelay", this.getMaxReconnectDelay()));
      this.setMsgTimeToLive(
          config.getIntProperty("producer.msgTimeToLive", this.getMsgTimeToLive()));
      this.setEventMaxSize(config.getIntProperty("producer.eventMaxSize", this.getEventMaxSize()));
    } else {

      InputStream in = Thread.currentThread().getContextClassLoader()
          .getResourceAsStream(this.databusConfig.CONFIG_LOCAL_DIR + "producer.properties");
      Properties properties = new Properties();
      properties.load(in);
      this.setStartupMaxReconnectAttempts(
          Integer.valueOf(properties.getProperty("producer.startupMaxReconnectAttempts")));
      this.setMaxReconnectAttempts(
          Integer.valueOf(properties.getProperty("producer.maxReconnectAttempts")));
      this.setOtherURLConfig(properties.getProperty("producer.otherURLConfig"));
      this.setTimeout(Integer.valueOf(properties.getProperty("producer.timeout")));
      this.setTransactedSession(
          "true".equals(properties.getProperty("producer.transactedSession")));
      this.setMaxReconnectDelay(
          Integer.valueOf(properties.getProperty("producer.maxReconnectDelay")));
      if (null != properties.getProperty("producer.msgTimeToLive")) {
        this.setMsgTimeToLive(Integer.valueOf(properties.getProperty("producer.msgTimeToLive")));
      }
      if (null != properties.getProperty("producer.eventMaxSize")) {
        this.setEventMaxSize(Integer.valueOf(properties.getProperty("producer.eventMaxSize")));
      }
    }

  }


  int getTimeout() {
    return timeout;
  }

  void setTimeout(int timeout) {
    this.timeout = timeout;
  }

  int getMaxReconnectDelay() {
    return maxReconnectDelay;
  }

  void setMaxReconnectDelay(int maxReconnectDelay) {
    this.maxReconnectDelay = maxReconnectDelay;
  }

  int getMaxReconnectAttempts() {
    return maxReconnectAttempts;
  }

  void setMaxReconnectAttempts(int maxReconnectAttempts) {
    this.maxReconnectAttempts = maxReconnectAttempts;
  }

  String getOtherURLConfig() {
    return otherURLConfig;
  }

  void setOtherURLConfig(String otherURLConfig) {
    this.otherURLConfig = otherURLConfig;
  }

  boolean isTransactedSession() {
    return transactedSession;
  }

  void setTransactedSession(boolean transactedSession) {
    this.transactedSession = transactedSession;
  }

  int getAcknowledgeMode() {
    return acknowledgeMode;
  }

  void setAcknowledgeMode(int acknowledgeMode) {
    this.acknowledgeMode = acknowledgeMode;
  }

  int getStartupMaxReconnectAttempts() {
    return startupMaxReconnectAttempts;
  }

  /**
   * 获取消息保留天数
   */
  int getMsgTimeToLive() {
    return msgTimeToLive;
  }

  void setMsgTimeToLive(int msgTimeToLive) {
    this.msgTimeToLive = msgTimeToLive;
  }

  /**
   * 允许发送消息的最大值，单位KB.若批量发送，则表示所有的事件的大小之和
   * @return
   */
  int getEventMaxSize() {
    return eventMaxSize;
  }

  void setEventMaxSize(int eventMaxSize) {
    this.eventMaxSize = eventMaxSize;
  }

  /**
   * 设置客户端启动时尝试连接服务器的最大次数，默认为5
   *
   * @param startupMaxReconnectAttempts 客户端启动时尝试连接服务器的最大次数
   */
  public void setStartupMaxReconnectAttempts(int startupMaxReconnectAttempts) {
    this.startupMaxReconnectAttempts = startupMaxReconnectAttempts;
  }

  /**
   * 将生产者客户端配置转换为URL参数
   */
  public String transform2URLConfig() {
    StringBuilder brokerURLConfig = new StringBuilder();
    brokerURLConfig.append("maxReconnectDelay=" + this.getMaxReconnectDelay());
    brokerURLConfig.append("&maxReconnectAttempts=" + this.getMaxReconnectAttempts());
    brokerURLConfig.append(this.getOtherURLConfig());
    return brokerURLConfig.toString();
  }
}
